package com.tmsb.sink;

import com.tmsb.pojo.ESScore;
import com.tmsb.pojo.TaimuMainUserStockBuy;
import com.tmsb.utils.*;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

/**
 * Description: 用taimu_main的user_stock_buy来获取折扣兑换绩效积分数量和折扣两个实时指标的sink
 * Version:1.0.0
 *
 * @author xiekb
 * @date 2020/7/4
 */
public class UserStockBuySink extends RichSinkFunction<TaimuMainUserStockBuy> implements CheckpointedFunction {

    private ListState<BigDecimal> allAmountValueState;
    private ListState<BigDecimal> allDiscountValueState;
    private ListState<Long> dateState;
    private BigDecimal allAmountValue;
    private BigDecimal allDiscountValue;
    private Long dateValue;
    private JedisCluster jedisCluster;
    private SimpleDateFormat sdf;




    @Override
    public void invoke(TaimuMainUserStockBuy value, Context context) throws Exception {
        //每天0时清空状态

        long eventTime = value.getDate();
        if(eventTime>= dateValue){
            allAmountValue = new BigDecimal(0);
            allDiscountValue = new BigDecimal(0);
            dateValue = Util.getTomorrowZeroPointTimestamps();
        }else if(eventTime< Util.getZeroPointTimestamps()){
            return;
        }

        //每日总贡献值
        BigDecimal amount = value.getAmount();
        allAmountValue = allAmountValue.add(amount);
        allDiscountValue = allDiscountValue.add(amount.multiply(value.getPrice()));

        //向redis发送数据
        String format = sdf.format(new Date(eventTime));
        jedisCluster.hset(ConfUtil.getUserStockBuyAmountRedisKey(),format,allAmountValue+"");
        jedisCluster.hset(ConfUtil.getUserStockBuyDiscountRedisKey(),format,allDiscountValue+"");
    }



    @Override
    public void close() throws Exception {
        super.close();
        jedisCluster.close();
    }


    //当请求 checkpoint 快照时，将调用此方法
    @Override
    public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
        allAmountValueState.clear();
        allDiscountValueState.clear();
        dateState.clear();
        allAmountValueState.add(allAmountValue);
        allDiscountValueState.add(allDiscountValue);
        dateState.add(dateValue);
    }

    //在分布式执行期间创建并行功能实例时，将调用此方法。
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        jedisCluster = RedisCluster.getInstance(ConfUtil.getRedisCluster()).getRedisClusterList().get(0);
        allAmountValue = new BigDecimal(0);
        allDiscountValue = new BigDecimal(0);
        dateValue = 0L;
        sdf = new SimpleDateFormat("yyyyMMdd");

        ListStateDescriptor<BigDecimal> allAmountStateDes =
                new ListStateDescriptor<>("allAmountState", TypeInformation.of(new TypeHint<BigDecimal>() {}));
        ListStateDescriptor<BigDecimal> allDiscountStateDes =
                new ListStateDescriptor<>("allDiscountState", TypeInformation.of(new TypeHint<BigDecimal>() {}));
        ListStateDescriptor<Long> dateStateDes =
                new ListStateDescriptor<>("taimu_main_date", TypeInformation.of(new TypeHint<Long>() {}));

        allAmountValueState = context.getOperatorStateStore().getListState(allAmountStateDes);
        allDiscountValueState = context.getOperatorStateStore().getListState(allDiscountStateDes);
        dateState = context.getOperatorStateStore().getListState(dateStateDes);

        if (context.isRestored()) {
            allAmountValue = allAmountValueState.get().iterator().next();
            allDiscountValue = allDiscountValueState.get().iterator().next();
            dateValue = dateState.get().iterator().next();
        }
    }
}
